package com.bw.gmall.realtime.app.dws;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

import org.apache.commons.beanutils.BeanUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.bw.gmall.realtime.bean.GoodsTrafficStats;
import com.bw.gmall.realtime.utils.*;

/**
 * 商品流量来源TOP10统计应用
 * 统计访客数最高的前10款商品，并分析其流量来源
 * - GoodsTrafficTopNApp.java ：商品流量TOP N统计
 * -
 */
/**
 * 流量域商品访问TopN应用
 * 负责统计商品访问量TopN数据，分析用户对商品的关注度
 * 主要功能点：
 * - 从Kafka读取页面日志数据
 * - 使用Flink水印策略和窗口处理
 * - 统计各个商品的访问量
 * - 计算TopN热门商品
 * - 使用Druid连接池进行数据存储或查询
 * 
 * 数据流向：
 * 页面日志数据 -> GoodsTrafficTopNApp -> 处理后输出热门商品TopN结果
 */
public class GoodsTrafficTopNApp {
    private static final Logger log = LoggerFactory.getLogger(GoodsTrafficTopNApp.class);

    public static void main(String[] args) throws Exception {
        // 先测试ClickHouse连接是否正常
        boolean clickHouseConnected = testClickHouseConnection();
        if (!clickHouseConnected) {
            log.error("ClickHouse连接测试失败，请检查配置和服务状态！");
            return;
        }
        
        // 如果连接正常，生成一些模拟数据插入到ClickHouse中进行测试
        generateTestData();
        
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置并行度，测试环境设为1
        env.setParallelism(1);

        // 2. 从Kafka读取页面日志数据
        String pageTopic = "ods_traffic";
        String groupId = "goods_traffic_topn_group";
        // 测试环境可以使用一个测试主题
        // String pageTopic = "test_page_log";
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId);
        SingleOutputStreamOperator<JSONObject> jsonObjStream = env.addSource(kafkaSource).name("Kafka数据源")
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        return JSON.parseObject(s);
                    }
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>
                                forBoundedOutOfOrderness(Duration.ofSeconds(10))
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts");
                                    }
                                })
                );

        // 3. 过滤商品详情页访问
        SingleOutputStreamOperator<JSONObject> goodsPageStream = jsonObjStream.filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                JSONObject pageObj = jsonObject.getJSONObject("page");
                boolean isGoodDetail = pageObj != null && "good_detail".equals(pageObj.getString("page_id"));
                if (isGoodDetail) {
                    log.info("过滤到商品详情页访问: {}", jsonObject);
                }
                return isGoodDetail;
            }
        }).returns(jsonObjStream.getType());

        // 4. 提取商品ID和流量来源信息
        SingleOutputStreamOperator<GoodsTrafficStats> goodsStatsStream = goodsPageStream.map(new MapFunction<JSONObject, GoodsTrafficStats>() {
            @Override
            public GoodsTrafficStats map(JSONObject jsonObject) throws Exception {
                String statsTime = DateFormatUtil.toYmdHms(jsonObject.getLong("ts"));                String statsType = "10s";
                
                // 获取商品ID，添加空值检查
                String goodsId = "unknown";
                JSONObject pageObj = jsonObject.getJSONObject("page");
                if (pageObj != null && pageObj.containsKey("item")) {
                    goodsId = pageObj.getString("item");
                }
                
                // 获取来源信息，添加空值检查
                String entry = "unknown";
                if (pageObj != null && pageObj.containsKey("entry")) {
                    entry = pageObj.getString("entry");
                }
                String sourceType = "unknown";
                if (entry.contains("goods")) {
                    sourceType = "商品流量";
                } else if (entry.equals("home") || entry.contains("category")) {
                    sourceType = "店铺流量";
                } else if (entry.contains("live")) {
                    sourceType = "直播流量";
                } else if (entry.contains("article")) {
                    sourceType = "内容流量";
                }
                
                // 初始化统计指标
                GoodsTrafficStats stats = new GoodsTrafficStats();
                stats.setStatsTime(statsTime);
                stats.setStatsType(statsType);
                stats.setGoodsId(goodsId);
                stats.setGoodsName("商品" + goodsId); // 临时名称，后续会关联维度表获取真实名称
                stats.setUvCount(1L); // 每个mid算作一个访客
                stats.setPvCount(1L); // 每个页面访问算作一个PV
                
                // 从actions字段中提取加购、收藏和支付行为
                long cartAddCount = 0L;
                long favoritesCount = 0L;
                long paymentUserCount = 0L;
                BigDecimal paymentAmount = BigDecimal.ZERO;
                
                if (jsonObject.containsKey("actions")) {
                    for (Object action : jsonObject.getJSONArray("actions")) {
                        JSONObject actionObj = (JSONObject) action;
                        String actionType = actionObj.getString("action").toLowerCase();
                        if ("addcart".equals(actionType)) {
                            cartAddCount = 1L;
                        } else if ("favor".equals(actionType)) {
                            favoritesCount = 1L;
                        } else if ("payment".equals(actionType)) {
                            paymentUserCount = 1L;
                            paymentAmount = new BigDecimal(100); // 示例值，实际应从订单数据获取
                        }
                    }
                }
                
                stats.setCartAddCount(cartAddCount);
                stats.setFavoritesCount(favoritesCount);
                stats.setPaymentUserCount(paymentUserCount);
                stats.setPaymentAmount(paymentAmount);
                stats.setSourceType(sourceType);
                stats.setCreateTime(new Date());
                
                return stats;
            }
        });

        // 5. 异步关联商品维度表，获取商品名称
        SingleOutputStreamOperator<GoodsTrafficStats> enrichedStatsStream = AsyncDataStream.unorderedWait(
                goodsStatsStream,
                new AsyncDimFunction(),
                60, TimeUnit.SECONDS,
                100
        );

        // 6. 按商品ID和统计时间分组，统计指标
        SingleOutputStreamOperator<GoodsTrafficStats> aggregatedStream = enrichedStatsStream
                .keyBy(new KeySelector<GoodsTrafficStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(GoodsTrafficStats stats) throws Exception {
                        // 按统计时间、统计类型、商品ID和来源类型分组
                        return new Tuple4<>(stats.getStatsTime(), stats.getStatsType(), 
                                           stats.getGoodsId(), stats.getSourceType());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .reduce(new ReduceFunction<GoodsTrafficStats>() {
                    @Override
                    public GoodsTrafficStats reduce(GoodsTrafficStats value1, GoodsTrafficStats value2) throws Exception {
                        // 累加各项指标
                        value1.setUvCount(value1.getUvCount() + value2.getUvCount());
                        value1.setPvCount(value1.getPvCount() + value2.getPvCount());
                        value1.setCartAddCount(value1.getCartAddCount() + value2.getCartAddCount());
                        value1.setFavoritesCount(value1.getFavoritesCount() + value2.getFavoritesCount());
                        value1.setPaymentUserCount(value1.getPaymentUserCount() + value2.getPaymentUserCount());
                        value1.setPaymentAmount(value1.getPaymentAmount().add(value2.getPaymentAmount()));
                        return value1;
                    }
                });

        // 7. 按统计时间和统计类型分组，计算商品TOP10
        SingleOutputStreamOperator<GoodsTrafficStats> top10Stream = aggregatedStream
                .keyBy(new KeySelector<GoodsTrafficStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(GoodsTrafficStats stats) throws Exception {
                        // 按统计时间、统计类型和来源类型分组，对每个类型计算TOP10商品
                        return new Tuple4<>(stats.getStatsTime(), stats.getStatsType(), 
                                           stats.getSourceType(), "goods_topn");
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10))) // 测试环境临时改为10秒窗口
                .apply(new WindowFunction<GoodsTrafficStats, GoodsTrafficStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> key, TimeWindow window, 
                                     Iterable<GoodsTrafficStats> input, Collector<GoodsTrafficStats> out) throws Exception {
                        // 将窗口内的数据转换为列表
                        List<GoodsTrafficStats> statsList = new ArrayList<>();
                        for (GoodsTrafficStats stats : input) {
                            // 创建副本以避免修改原始数据
                    GoodsTrafficStats copy = new GoodsTrafficStats();
                    try {
                        BeanUtils.copyProperties(copy, stats);
                    } catch (Exception e) {
                        log.error("Copy properties error: " + e.getMessage(), e);
                        // 如果复制失败，使用原始对象
                        copy = stats;
                    }
                            statsList.add(copy);
                        }

                        // 按访客数降序排序，取TOP10
                        List<GoodsTrafficStats> top10List = statsList.stream()
                                .sorted((s1, s2) -> s2.getUvCount().compareTo(s1.getUvCount()))
                                .limit(10)
                                .collect(Collectors.toList());

                        // 为TOP10数据添加排名
                        for (int i = 0; i < top10List.size(); i++) {
                            GoodsTrafficStats stats = top10List.get(i);
                            stats.setRank(i + 1);
                            out.collect(stats);
                        }
                        log.info("窗口结束，计算出TOP10商品数量: {}", top10List.size());
                    }
                }).name("top10计算");

        // 8. 数据校验和转换，确保写入ClickHouse的数据格式正确
        SingleOutputStreamOperator<GoodsTrafficStats> validatedStream = top10Stream.map(new MapFunction<GoodsTrafficStats, GoodsTrafficStats>() {
            @Override
            public GoodsTrafficStats map(GoodsTrafficStats stats) throws Exception {
                // 确保排名不为null
                if (stats.getRank() == null) {
                    stats.setRank(0);
                }
                // 确保所有统计字段不为null
                if (stats.getUvCount() == null) stats.setUvCount(0L);
                if (stats.getPvCount() == null) stats.setPvCount(0L);
                if (stats.getCartAddCount() == null) stats.setCartAddCount(0L);
                if (stats.getFavoritesCount() == null) stats.setFavoritesCount(0L);
                if (stats.getPaymentUserCount() == null) stats.setPaymentUserCount(0L);
                if (stats.getPaymentAmount() == null) stats.setPaymentAmount(BigDecimal.ZERO);
                // 确保来源类型不为null
                if (stats.getSourceType() == null || stats.getSourceType().isEmpty()) {
                    stats.setSourceType("unknown");
                }
                log.info("数据准备写入ClickHouse: {}", stats);
                return stats;
            }
        }).name("数据校验");
        // 9. 将结果写入ClickHouse
        validatedStream.addSink(MyClickHouseUtil.getSinkFunction(
                "insert into dws_goods_traffic_topn_window(stats_time, stats_type, goods_id, goods_name, uv_count, pv_count, cart_add_count, favorites_count, payment_user_count, payment_amount, source_type, rank) values(?,?,?,?,?,?,?,?,?,?,?,?)"
        ));

        // 9. 执行作业
        env.execute("GoodsTrafficTopNApp");
    }

    /**
     * 测试ClickHouse连接是否正常
     */
    private static boolean testClickHouseConnection() {
        Connection conn = null;
        try {
            log.info("开始测试ClickHouse连接...");
            
            // 使用JDBC直接测试连接
            Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
            conn = DriverManager.getConnection("jdbc:clickhouse://hadoop102:8123/gmall");
            
            // 检查连接是否有效
            boolean isValid = conn.isValid(5); // 5秒超时
            log.info("ClickHouse连接有效性: {}", isValid);
            
            // 如果连接有效，检查数据库和表是否存在
            if (isValid) {
                checkDBAndTable(conn);
            }
            
            return isValid;
        } catch (Exception e) {
            log.error("ClickHouse连接测试失败: {}", e.getMessage(), e);
            return false;
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    log.error("关闭ClickHouse连接时出错: {}", e.getMessage());
                }
            }
        }
    }
    
    /**
     * 检查数据库和表是否存在
     */
    private static void checkDBAndTable(Connection conn) throws SQLException {
        try (PreparedStatement stmt = conn.prepareStatement("SHOW DATABASES LIKE 'gmall'")) {
            try (ResultSet rs = stmt.executeQuery()) {
                if (rs.next()) {
                    log.info("gmall数据库存在");
                } else {
                    log.error("gmall数据库不存在！");
                }
            }
        }
        
        try (PreparedStatement stmt = conn.prepareStatement("SHOW TABLES LIKE 'dws_goods_traffic_topn_window'")) {
            try (ResultSet rs = stmt.executeQuery()) {
                if (rs.next()) {
                    log.info("dws_goods_traffic_topn_window表存在");
                    // 查询表结构
                    try (PreparedStatement metaStmt = conn.prepareStatement("DESCRIBE TABLE dws_goods_traffic_topn_window")) {
                        try (ResultSet metaRs = metaStmt.executeQuery()) {
                            log.info("表结构:");
                            while (metaRs.next()) {
                                String name = metaRs.getString(1);
                                String type = metaRs.getString(2);
                                log.info("  {}: {}", name, type);
                            }
                        }
                    }
                    
                    // 查询当前记录数
                    try (PreparedStatement countStmt = conn.prepareStatement("SELECT COUNT(*) FROM dws_goods_traffic_topn_window")) {
                        try (ResultSet countRs = countStmt.executeQuery()) {
                            if (countRs.next()) {
                                long count = countRs.getLong(1);
                                log.info("表中现有记录数: {}", count);
                            }
                        }
                    }
                } else {
                    log.error("dws_goods_traffic_topn_window表不存在！");
                }
            }
        }
    }
    
    /**
     * 生成测试数据并直接插入到ClickHouse中
     */
    private static void generateTestData() {
        Connection conn = null;
        PreparedStatement stmt = null;
        
        try {
            Class.forName("ru.yandex.clickhouse.ClickHouseDriver");
            conn = DriverManager.getConnection("jdbc:clickhouse://hadoop102:8123/gmall");
            
            // 生成10条模拟数据
            String sql = "INSERT INTO dws_goods_traffic_topn_window (stats_time, stats_type, goods_id, goods_name, uv_count, pv_count, cart_add_count, favorites_count, payment_user_count, payment_amount, source_type, rank) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
            stmt = conn.prepareStatement(sql);
            
            log.info("开始生成模拟测试数据...");
            
            for (int i = 1; i <= 10; i++) {
                // 清除之前的参数设置
                stmt.clearParameters();
                
                // 设置参数
                stmt.setString(1, "2024-10-30 10:00:00"); // stats_time
                stmt.setString(2, "10s"); // stats_type
                stmt.setString(3, "1000" + i); // goods_id
                stmt.setString(4, "测试商品" + i); // goods_name
                stmt.setLong(5, 100L - i); // uv_count (故意设置不同的值用于排序)
                stmt.setLong(6, 200L - i * 2); // pv_count
                stmt.setLong(7, 50L - i); // cart_add_count
                stmt.setLong(8, 30L - i); // favorites_count
                stmt.setLong(9, 20L - i); // payment_user_count
                stmt.setBigDecimal(10, new BigDecimal(1000.0 - i * 10)); // payment_amount
                stmt.setString(11, i % 2 == 0 ? "app" : "web"); // source_type
                stmt.setInt(12, i); // rank
                
                // 执行插入
                stmt.addBatch();
                
                log.info("生成测试数据: goods_id=1000{}, uv_count={}", i, 100L - i);
            }
            
            // 执行批处理
            int[] rowsAffected = stmt.executeBatch();
            log.info("成功插入测试数据，影响行数: {}", rowsAffected.length);
            
            // 验证插入结果
            try (PreparedStatement checkStmt = conn.prepareStatement("SELECT COUNT(*) FROM dws_goods_traffic_topn_window WHERE stats_time = '2024-10-30 10:00:00'")) {
                try (ResultSet rs = checkStmt.executeQuery()) {
                    if (rs.next()) {
                        long count = rs.getLong(1);
                        log.info("测试数据验证: 时间点'2024-10-30 10:00:00'的记录数为: {}", count);
                    }
                }
            }
            
        } catch (Exception e) {
            log.error("生成或插入测试数据时出错: {}", e.getMessage(), e);
        } finally {
            if (stmt != null) {
                try { stmt.close(); } catch (SQLException e) {}
            }
            if (conn != null) {
                try { conn.close(); } catch (SQLException e) {}
            }
        }
    }
    
    /**
     * 异步查询商品维度信息
     * 优化：使用Flink推荐的异步处理模式，不再创建自定义线程
     */
    public static class AsyncDimFunction extends RichAsyncFunction<GoodsTrafficStats, GoodsTrafficStats> {
        private DruidDataSource dataSource = null;

        @Override
        public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
            super.open(parameters);
            // 使用正确的数据源配置
            dataSource = DruidDSUtil.createDataSource();
        }

        @Override
        public void asyncInvoke(GoodsTrafficStats stats, ResultFuture<GoodsTrafficStats> resultFuture) throws Exception {
            // 不再创建新线程，使用Flink提供的线程池
            DruidPooledConnection connection = null;
            try {
                connection = dataSource.getConnection();
                // 查询商品维度信息
                JSONObject dimInfo = DimUtil.getDimInfo(connection, "DIM_GOODS_INFO", stats.getGoodsId());
                if (dimInfo != null && dimInfo.size() > 0) {
                    // 更新商品名称
                    if (dimInfo.containsKey("goods_name")) {
                        stats.setGoodsName(dimInfo.getString("goods_name"));
                    }
                }
                resultFuture.complete(Collections.singletonList(stats));
            } catch (Exception e) {
                // 改进异常处理，避免在生产环境打印完整堆栈
                System.err.println("异步查询商品维度信息失败: " + e.getMessage());
                // 即使失败也返回原始数据，确保流程继续
                resultFuture.complete(Collections.singletonList(stats));
            } finally {
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        @Override
        public void close() throws Exception {
            super.close();
            if (dataSource != null) {
                dataSource.close();
            }
        }
    }
}

